package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.bean.PaymentInfo;
import com.atguigu.realtime.bean.PaymentWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Map;

import static com.atguigu.realtime.common.Constant.TOPIC_DWD_PAYMENT_INFO;
import static com.atguigu.realtime.common.Constant.TOPIC_DWM_ORDER_WIDE;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/15 9:45
 */
public class DwmPaymentWideApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmPaymentWideApp().init(3004, 1, "DwmPaymentWideApp", "DwmPaymentWideApp",
                                     TOPIC_DWD_PAYMENT_INFO, TOPIC_DWM_ORDER_WIDE);
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> topicStreamMap) {
        KeyedStream<PaymentInfo, Long> paymentInfoStream = topicStreamMap
            .get(TOPIC_DWD_PAYMENT_INFO)
            .map(json -> JSON.parseObject(json, PaymentInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((info, ts) -> info.getTs())
            )
            .keyBy(PaymentInfo::getOrder_id);
    
    
        KeyedStream<OrderWide, Long> orderWideStream = topicStreamMap
            .get(TOPIC_DWM_ORDER_WIDE)
            .map(json -> JSON.parseObject(json, OrderWide.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((wide, ts) -> AtguiguUtil.toTs(wide.getCreate_time()))
            )
            .keyBy(OrderWide::getOrder_id);
    
        paymentInfoStream
            .intervalJoin(orderWideStream)
            .between(Time.minutes(-45), Time.seconds(5))
            .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                @Override
                public void processElement(PaymentInfo left,
                                           OrderWide right,
                                           Context ctx,
                                           Collector<PaymentWide> out) throws Exception {
                    out.collect(new PaymentWide(left, right));
                }
            })
            .map(JSON::toJSONString)
            .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWM_PAYMENT_WIDE));
    }
}
/*
支付宽表思路1:
    和订单宽表完全一样, 让支付表和order_detail进行 join, 然后补齐需要的维度信息

    注意的地方:
        order_detail和order_info他们几乎是同时产生, 下单的时候生成

        支付表一般会比较晚于订单表的生成

        所以让支付表和order_detail join的这个下界要比较小, 对内存的压力比较大


支付宽表的思路2:
    
    来了一条支付表数据,
    订单1支付
              详情1
              详情2

   把详情表当做一个维度表数据, 存储到hbase(Phoenix)中, 当订单1的支付来了, 根据订单id1 去hbase中查找对应的详情, 然后join到一起

   然后再去补充维度数据...

支付宽表的思路3:
    不想再去join维度数据, 但是还需要维度数据.
    
    dwm订单宽表已经是商品粒度, 而且已经有了维度信息
    能不能让支付表和订单宽表进行双流join, 省去了单独去join维度的步骤
 */